#include "ServerConncetionLibraryImpl.h"

#include <boost/asio/placeholders.hpp>

#define NUMBER_OF_COMPLETION_THREADS 2

namespace AlgoTrading
{
	namespace ConnectionLibrary
	{

		ServerConnectionLibraryImpl::ServerConnectionLibraryImpl(ClientConnectionAcceptorPtr acceptor, const std::string& address, const std::string& port)
			: connectionAcceptor(acceptor)
			, io_service(new boost::asio::io_service())
			, acceptor(*io_service)
			, address(address)
			, port(port)
			, newConnectionSocket(new boost::asio::ip::tcp::socket(*io_service))
			, started(false)
		{
		}

		ServerConnectionLibraryImpl::~ServerConnectionLibraryImpl(void)
		{
			Stop();
		}

		void ServerConnectionLibraryImpl::Start()
		{
			if (!started)
			{
				for(int i =0; i < NUMBER_OF_COMPLETION_THREADS; ++i)
				{
					threads.push_back(std::tr1::shared_ptr<boost::thread>(new boost::thread(boost::bind(&ServerConnectionLibraryImpl::ProcessingCycle, this))));
				}
				// Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).
				boost::asio::ip::tcp::resolver resolver(*io_service);
				boost::asio::ip::tcp::resolver::query query(address, port);
				boost::asio::ip::tcp::endpoint endpoint = *resolver.resolve(query);
				acceptor.open(endpoint.protocol());
				acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
				acceptor.bind(endpoint);
				acceptor.listen();
				acceptor.async_accept(*newConnectionSocket, boost::bind(&ServerConnectionLibraryImpl::HandleAccept, this, boost::asio::placeholders::error));
			}
		}

		void ServerConnectionLibraryImpl::ProcessingCycle()
		{
			boost::asio::io_service::work work(*io_service);
			io_service->run();
		}

		void ServerConnectionLibraryImpl::HandleAccept(const boost::system::error_code& error)
		{
			if (!error)
			{
				newConnectionSocket->set_option(boost::asio::ip::tcp::no_delay(true));

				ClientConnectionHandlerPtr clientConnectionHandler(new ClientConnectionHandler(*io_service, newConnectionSocket, connectionAcceptor, connectionManager));
				std::tr1::weak_ptr<ClientNotificationProcessor> clientNotificationProcessor(std::tr1::static_pointer_cast<ClientNotificationProcessor>(clientConnectionHandler));

				ClientRequestProcessorPtr requestProcessor = connectionAcceptor->OnClientConnected(clientNotificationProcessor);
				clientConnectionHandler->SetRequestProcessor(requestProcessor);

				connectionManager.Start(clientConnectionHandler);
				newConnectionSocket.reset(new boost::asio::ip::tcp::socket(*io_service)); 

				acceptor.async_accept(*newConnectionSocket, boost::bind(&ServerConnectionLibraryImpl::HandleAccept, this, boost::asio::placeholders::error));
			}
		}

		void ServerConnectionLibraryImpl::Stop()
		{
			if (started)
			{
				io_service->post(boost::bind(&ServerConnectionLibraryImpl::HandleStop, this));

				//wait for threads to finish
				for(ThreadsList::iterator it = threads.begin(); it != threads.end(); ++it)
				{
					(*it)->join();
				}
				threads.clear();

				started = false;
			}
		}

		void ServerConnectionLibraryImpl::HandleStop()
		{
			connectionManager.StopAll();

			io_service->stop();
		}

	}
}
